package com.remoter.basic.consumer.support;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.remoter.basic.bean.support.ConsumerModule;
import com.remoter.basic.bean.support.ProviderModule;
import com.remoter.basic.bean.support.RemoterRequest;
import com.remoter.basic.configure.IConfiguration;
import com.remoter.basic.configure.support.AbstractConfiguration;
import com.remoter.basic.consumer.IConsumer;
import com.remoter.basic.registry.IRegistry;
import com.remoter.basic.registry.IRegistryProviderListener;
import com.remoter.basic.transport.IClient;
import com.remoter.basic.util.AbortPolicyWithReport;
import com.remoter.basic.util.Final;
import com.remoter.basic.util.NamedThreadFactory;
import com.remoter.basic.util.ServiceLoader;
import com.remoter.basic.util.StringUtil;

/**
 * @author			koko
 * @date 			2017-08-10 14:33:12
 * @description 	类功能说明
 */
public abstract class AbstractConsumer implements IConsumer{
	
	protected static final Logger logger = LoggerFactory.getLogger(AbstractConsumer.class);
	
	protected ConcurrentHashMap<String,IClient> clients = new ConcurrentHashMap<String,IClient>();
	protected ConcurrentHashMap<String,ConsumerModule> modules = new ConcurrentHashMap<String,ConsumerModule>();
	protected ConcurrentHashMap<String,ProviderModule> services = new ConcurrentHashMap<String,ProviderModule>();
	
	protected IConfiguration configuration;
	protected IRegistry registry;
	protected ExecutorService executorService;
	
	public AbstractConsumer(){
		this.configuration = AbstractConfiguration.getConfiguration();
		this.registry = ServiceLoader.getService(IRegistry.class,this.configuration.getOption(Final.O_VAL_REGISTRY));
		String taskName = "CLIENT_TASK";
		executorService = new ThreadPoolExecutor(
				Final.PROCRESS,Final.PROCRESS*100,
				1000,
				TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(Final.PROCRESS*500),
                new NamedThreadFactory(taskName,true),new AbortPolicyWithReport(taskName));
		this.initListeners();
		this.initServices();
	}
	
	@Override
	public void close() throws IOException {
		for(Entry<String,ConsumerModule> entry : modules.entrySet()){
			ConsumerModule module = entry.getValue();
			if(this.registry.detachConsumer(module)){
				modules.remove(entry.getKey());
			}else{
				logger.error("detach consumer error for key : " + module.getKey());
			}
		}
		for(Entry<String,IClient> entry : clients.entrySet()){
			try{
				entry.getValue().disConnect();
			}catch(Exception e){
				logger.error("disconnect client error for key :" + entry.getKey());
			}
		}
		modules.clear();
		services.clear();
		clients.clear();
	}

	@Override
	public void attachConsumer(ConsumerModule consumerModule) {
		String key = consumerModule.getKey();
		if(modules.containsKey(key)){
			return;
		}
		if(this.registry.attachConsumer(consumerModule)){
			try{
				consumerModule.setProxy(this.createProxy(consumerModule));
				this.modules.putIfAbsent(key,consumerModule);
			}catch(Exception e){
				logger.error(e.getMessage(),e);
				this.registry.detachConsumer(consumerModule);
				this.modules.remove(key);
			}
		}
	}

	@Override
	public void detachConsumer(ConsumerModule consumerModule) {
		String key = consumerModule.getKey();
		if(!modules.containsKey(key)){
			return;
		}
		if(this.registry.detachConsumer(consumerModule)){
			this.modules.remove(key);
		}
	}

	@Override
	public ConsumerModule getConsumer(RemoterRequest remoterRequest){
		for(Entry<String,ConsumerModule> entry : modules.entrySet()){
			ConsumerModule module = entry.getValue();
			if(
				StringUtil.isEquals(module.getProtocol(),remoterRequest.getProtocol()) && 
				StringUtil.isEquals(module.getType(), remoterRequest.getType()) && 
				StringUtil.isEquals(module.getBean(), remoterRequest.getBean())
			){
				return module;
			}
		}
		return null;
	}

	@Override
	public List<ProviderModule> getProviders(ConsumerModule consumerModule){
		List<ProviderModule> result = new ArrayList<ProviderModule>();
		for(Entry<String,ProviderModule> entry : services.entrySet()){
			ProviderModule module = entry.getValue();
			if(
				StringUtil.isEquals(module.getProtocol(),consumerModule.getProtocol()) && 
				StringUtil.isEquals(module.getType(),consumerModule.getType()) && 
				StringUtil.isEquals(module.getBean(),consumerModule.getBean())
			){
				result.add(module);
			}
		}
		return result;
	}

	@Override
	public IClient getClient(ConsumerModule consumerModule, boolean autoConnect)throws Exception{
		String key = this.createKey(consumerModule);
		IClient client = this.clients.get(key);
		if(null == client){
			if(autoConnect){
				IClient tempClient = ServiceLoader.getService(IClient.class,consumerModule.getProtocol());
				if(null == tempClient){
					throw new IllegalStateException("protocol not found");
				}
				tempClient.connect(executorService,new InetSocketAddress(consumerModule.getRemoteHost(), consumerModule.getRemotePort()));
				InetSocketAddress address = tempClient.getLocalAddress();
				ConsumerModule newConsumerModule = consumerModule.cloneInstance();
				newConsumerModule.setHost(address.getHostString());
				newConsumerModule.setPort(address.getPort());
				this.clients.put(key,tempClient);
				client = tempClient;
				if(this.registry.updateConsumer(consumerModule,newConsumerModule)){
					modules.remove(consumerModule.getKey());
					modules.put(newConsumerModule.getKey(),newConsumerModule);
				}
			}
		}
		return client;
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public <T> T getConsumerProxy(Class<T> type, ConsumerModule consumerModule) {
		String key = consumerModule.getKey();
		if(this.modules.containsKey(key)){
			return (T)this.modules.get(key).getProxy();
		}
		return null;
	}

	protected abstract Object createProxy(ConsumerModule consumerModule)throws Exception;
	
	protected String createKey(ConsumerModule consumerModule){
		return String.format("%s-%s-%s",consumerModule.getProtocol(),consumerModule.getRemoteHost(),consumerModule.getRemotePort());
	}
	
	protected void initListeners(){
		this.registry.attachProviderListener(new IRegistryProviderListener(){
			@Override
			public void onUpdate(String path,ProviderModule data){
				services.putIfAbsent(data.getKey(),data);
			}
			@Override
			public void onDelete(String path, ProviderModule data){
				String key = data.getKey();
				if(services.containsKey(key)){
					services.remove(key);
				}
			}
			@Override
			public void onCreate(String path, ProviderModule data) {
				services.putIfAbsent(data.getKey(),data);
			}
		});
	}
	
	protected void initServices(){
		List<ProviderModule> values = this.registry.listProviders();
		if(null == values || values.size() == 0){
			return;
		}
		for(ProviderModule module : values){
			String key = module.getKey();
			services.putIfAbsent(key,module);
		}
	}
	
}